package com.sbp.message.consumer;

import com.alibaba.fastjson.JSON;
import com.dap.api.ISunriseService;
import com.dap.exception.BusinessException;
import com.dap.exception.PlatformException;
import com.dap.param.CommonServiceContext;
import com.dap.param.StringInput;
import com.dap.utils.TraceIdUtils;
import com.sbp.message.api.common.CommonUtils;
import com.sbp.message.api.entity.*;
import com.sbp.message.entity.ConsumerMessageRecord;
import com.sbp.message.api.service.MessageService;
import com.sbp.message.entity.SunriseServiceAndParam;
import com.sbp.message.exception.ExceptionConstants;
import com.sbp.message.service.ConsumerMessageRecordService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;

public class MessageConsumer {
    private static final Logger logger = LoggerFactory.getLogger(MessageConsumer.class);

    private MessageService messageService;
    public void setMessageService(MessageService messageService) {
        this.messageService = messageService;
    }


    private ConsumerMessageRecordService consumerMessageRecordService;
    public void setConsumerMessageRecordService(ConsumerMessageRecordService consumerMessageRecordService) {
        this.consumerMessageRecordService = consumerMessageRecordService;
    }

    private ServiceProcessor serviceProcessor;
    public void setServiceProcessor(ServiceProcessor serviceProcessor) {
        this.serviceProcessor = serviceProcessor;
    }

    private ConsumerServicesManager consumerServicesManager;
    public void setConsumerServicesManager(ConsumerServicesManager consumerServicesManager) {
        this.consumerServicesManager = consumerServicesManager;
    }



    // 消费之前, 判断是否需要消费
    private boolean beforeConsume(Message message, String group) {
        ConsumerMessageRecord consumerMessageRecord = consumerMessageRecordService.selectOne(message.getId(), group);

        // 没有本地消费记录, 可以消费
        if (null == consumerMessageRecord) {
            return true;
        }

        // 复活
        if (message.retry >= CommonUtils.MAX_RETIES
                && LocalRecordStatus.succeed.getValue() != consumerMessageRecord.getStatus()
                && LocalRecordStatus.processing.getValue() != consumerMessageRecord.getStatus()) {
            logger.info("retry message, consumerMessageRecord={}, message=\n{}", JSON.toJSONString(consumerMessageRecord), JSON.toJSONString(message));
            consumerMessageRecordService.delete(message.getId(), group);   // 删除本地业务异常记录
            return true;
        }

        return false;
    }
    private void consumerMessageInternal(
            Message message,
            String group,
            ISunriseService service, StringInput stringInput) throws Exception {
        Exception exception = null;
        try {
            boolean isNeedHandle = serviceProcessor.executeServiceAndInsertRecord(message, group, service, stringInput);
            if (!isNeedHandle) {
                return;
            }
        } catch (Exception e) {
            logger.warn("", e);
            exception = e;
        }

        // 尝试正常结束消息
        if (exception == null) {
            try {
                messageService.tryEndMessageWithConsumeSucceed(message, group);
            } catch (Exception e) {
                logger.error("严重错误,必须人工干预", e);
                throw new PlatformException(ExceptionConstants.MESSAGE_REMOTE_ERROR, e);
            }
        }
        // 处理异常
        else {
            handleException(message, group, exception);
        }
    }

    /**
     * 消费消息并处理异常
     */
    public void consumeMessage(String messageJson, String group) throws Exception {
        Message message;
        try {
            logger.debug("message={}", messageJson);
            message = JSON.parseObject(messageJson, Message.class);
        } catch (Exception e) {
            logger.warn("message content is invalid");
            throw e;
        }

        // 消费者保持和消息发送者统一的traceId
        TraceIdUtils.setTraceId(message.traceId);

        SunriseServiceAndParam ssap = consumerServicesManager.resolveSunriseServiceAndParam(message, group);
        if (null == ssap) return;

        // 消息已经处理过, 不再处理
        if (!beforeConsume(message, group)) {
            if (logger.isDebugEnabled()) {
                logger.debug("message already handled, message={}", JSON.toJSONString(message));
            }
            return;
        }

        CommonServiceContext.setGroupName(group);
        CommonServiceContext.setCtrlData(ssap.stringInput.getCtrlData());
        CommonServiceContext.setIsFrontRequest(false);
        try {
            consumerMessageInternal(message, group, ssap.service, ssap.stringInput);
        } finally {
            CommonServiceContext.removeGroupName();
            CommonServiceContext.removeCtrlData();
            CommonServiceContext.removeIsFrontRequest();
        }
    }


    /**
     * 处理异常
     */
    private void handleException(Message message, String group, Exception exception) throws Exception {
        // 业务异常, 尝试错误结束消息
        if (hasBusinessExceptionInCause(exception)) {
            try {
                serviceProcessor.insertConsumerMessageRecord(message, group, LocalRecordStatus.businessFailed);
                messageService.tryEndMessageWithBusinessFailed(
                        message, group,
                        exception.getMessage(), generateExceptionStackTraceStr(exception));
            } catch (Exception e) {
                logger.warn("", e);
                throw new PlatformException(ExceptionConstants.MESSAGE_REMOTE_ERROR, e);
            }
        }
        // 其他异常, 修改状态为发送中, 等待检查任务检查重发; 忽略sql执行可能出现的异常消息, 只打印; 最后抛出原来的异常
        else {
            try {
                final String errorLog = generateExceptionStackTraceStr(exception);
                logger.warn("{}", errorLog);
                messageService.updateMessageStatus(
                        message.getId(),
                        MessageStatus.sending,
                        exception.getMessage(),
                        errorLog);
            } catch (Exception e) {
                logger.warn("", e);
            }
            throw exception;
        }
    }

    private static boolean hasBusinessExceptionInCause(Throwable e) {
        while (e != null) {
            if (e instanceof BusinessException) {
                return true;
            }
            e = e.getCause();
        }
        return false;
    }
    private static String generateExceptionStackTraceStr(Throwable e) {
        ByteArrayOutputStream os = new ByteArrayOutputStream();
        PrintStream ps = new PrintStream(os);

        e.printStackTrace(ps);

        try {
            return os.toString("UTF-8");
        } catch (UnsupportedEncodingException e1) {
            return null;
        } finally {
            ps.close();
        }
    }

}